package com.cande.loki.jmh.aeron;

import io.aeron.Aeron;
import io.aeron.CommonContext;
import io.aeron.Publication;
import io.aeron.driver.MediaDriver;
import io.aeron.driver.ThreadingMode;
import org.agrona.concurrent.IdleStrategy;
import org.agrona.concurrent.SleepingIdleStrategy;
import org.agrona.concurrent.UnsafeBuffer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.ByteBuffer;
import java.util.Scanner;

public class Sender {

    private static final Logger logger = LoggerFactory.getLogger(Sender.class);

    public static final UnsafeBuffer unsafeBuffer = new UnsafeBuffer(ByteBuffer.allocateDirect(256));

    public static void main(String[] args) {

        final IdleStrategy idle = new SleepingIdleStrategy();

        final MediaDriver.Context mediaDriverCtx = new MediaDriver.Context()
                .aeronDirectoryName(CommonContext.getAeronDirectoryName() + "-server")
                .dirDeleteOnStart(true)
                .dirDeleteOnShutdown(true)
                .threadingMode(ThreadingMode.SHARED);


        MediaDriver driver = MediaDriver.launchEmbedded(mediaDriverCtx);
        final Aeron.Context aeronCtx = new Aeron.Context()
                .aeronDirectoryName(driver.aeronDirectoryName());

        Aeron aeron = Aeron.connect(aeronCtx);

        Publication pub = aeron.addExclusivePublication("aeron:udp?endpoint=127.0.0.1:2000", 11);


        while (true) {

            String scanner = scanner();
            logger.info("sending : " + scanner);
            if (!pub.isConnected()) {
                logger.info("没有消费者，不发送");
            } else {
                unsafeBuffer.putStringAscii(0,scanner);
                while (pub.offer(unsafeBuffer) < 0)
                {
                    idle.idle();
                }
            }
        }
    }

    private static String scanner() {
        Scanner scanner = new Scanner(System.in);
        if (scanner.hasNext()) {
            String ipt = scanner.next();
            if (ipt!=null && ipt.length()>0) {
                return ipt;
            }
        }
        throw new RuntimeException();
    }

}
